package com.huan.hotitems;

import com.huan.bean.ItemViewCount;
import com.huan.bean.UserBehavior;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

public class HotItems {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism( 1 );
        //设置事件时间语义
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );
        //获取数据路径
        String filePath = "E:\\Project\\UserBehaviorAnalysis_Java\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv";
        //将数据读入进来
        DataStream<String> inputStream = env.readTextFile( filePath );
        //转换为POJO，分配时间戳和watermark
        DataStream<UserBehavior> dataStream = inputStream.map( line -> {
            String[] fields = line.split( "," );
            return new UserBehavior( new Long( fields[0] ), new Long( fields[1] ), new Integer( fields[2] ), fields[3], new Long( fields[4] ) );
        } )
                //给一个watermarks
                .assignTimestampsAndWatermarks( new AscendingTimestampExtractor<UserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(UserBehavior element) {
                        return element.getTimestamp() * 1000L;
                    }
                } );


        DataStream<ItemViewCount> windowAggStream = dataStream
                .filter( data -> "pv".equals( data.getBehavior() ) ) //过滤掉PV行为
                .keyBy( "itemId" ) // 按照商品ID分组
                .timeWindow( Time.hours( 1 ), Time.minutes( 5 ) ) //给一个滑动窗口，每隔五分钟出来一次
                .aggregate( new ItemCountAgg(), new WindowItemCountResult() );


        //收集同一窗口的所有商品的count数据,排序输出TopN
        DataStream<String> resultStream = windowAggStream
                .keyBy( "windowEnd" ) //按照窗口分组
                .process( new TopHotltems( 5 ) ); //用自定义处理函数 ，排序前五

        //打印输出
        resultStream.print();

        env.execute( "HotItems" );
    }


}
